package breadbox.pastry.routing;

import java.io.Serializable;
import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import breadbox.pastry.Conf;
import breadbox.pastry.PastryNode;

/**
 * Wraps a Priority Queue of RoutingNodes which uses a comparator to sort them
 * based on their distance from the base.
 * @author Daniel
 *
 */
public class NeighborhoodSet implements Serializable {
	private static final long serialVersionUID = -6296867450433887612L;
	private PriorityQueue< RoutingNode > set_;
	private ReadWriteLock lock_;
	private RoutingNode base_;
	
	public NeighborhoodSet( NeighborhoodSet other ) {
		lock_ = new ReentrantReadWriteLock();
		set_ = new PriorityQueue< RoutingNode >( Conf.m(), new DistanceComparator() );
		
		if (other != null)
		{
			base_ = new RoutingNode(other.base_);
			RoutingNode[ ] nodes = other.toArray();
			
			if( nodes != null ) {
				for( RoutingNode node : nodes ) {
					add( node );
				}
			}
		}
	}
	
	public NeighborhoodSet( RoutingNode base ) {
		lock_ = new ReentrantReadWriteLock();
		set_ = new PriorityQueue< RoutingNode >( Conf.m( ), new DistanceComparator( ) );
		base_ = base;
		base_.setType( RoutingNode.RoutingNodeType.ROUTE );
	}
	
	public int size( ) {
		lock_.readLock().lock();
		int size = set_.size();
		lock_.readLock().unlock();
		
		return size;
	}
	
	public void clear( ) {
		lock_.writeLock().lock();
		set_.clear( );
		lock_.writeLock().unlock();
	}
	
	public boolean isEmpty( ) {
		lock_.readLock().lock();
		boolean empty = set_.isEmpty( );
		lock_.readLock().unlock();
		return empty;
	}
	
	public boolean remove( RoutingNode n ) {
		lock_.writeLock().lock();
		boolean ret = set_.remove( n );
		lock_.writeLock().unlock();
		return ret;
	}
	
	public RoutingNode peek( ) {
		lock_.readLock().lock();
		RoutingNode ret = null;
		
		if( !set_.isEmpty() )
			ret = new RoutingNode(set_.peek());

		lock_.readLock().unlock();
		return ret;
	}
	
	
	// TODO (Daniel) add calls to application.proximity each time, 
	// and filter out nodes that are too far when there are more than M nodes
	// in the set
	public boolean add( RoutingNode n ) {
		if( n == null || n.equals(base_) )
			return false;
		
		lock_.writeLock().lock();
		n.setDistance( PastryNode.getPastryNode().proximity(n) );
		n.setType( RoutingNode.RoutingNodeType.NEIGHBOR );
		
		boolean changed = false;
		if( !( !set_.isEmpty() && n.getDistance() >= set_.peek().getDistance() && set_.size() >= Conf.m() ) ) {
			changed = set_.add( n );
		}
		
		if( set_.size( ) > Conf.m( ) ) {
			set_.poll();
		}
		
		lock_.writeLock().unlock();
		return changed;
	}
	
	public RoutingNode[ ] toArray( ) {
		lock_.readLock().lock();
		
		RoutingNode[ ] r = new RoutingNode[ set_.size( ) ];
		r = set_.toArray( r );

		lock_.readLock().unlock();
		return r;
	}

	/**
	 * Puts farther away nodes first, so that we have the max at the head of a priority queue.
	 * @author Daniel
	 *
	 */
	private class DistanceComparator implements Comparator< RoutingNode >, Serializable {
		private static final long serialVersionUID = -995741187581832192L;

		public int compare( RoutingNode o1, RoutingNode o2 ) {
			return o2.getDistance( ) - o1.getDistance( );
		}
	}
}
